package xyz.xtt.schedule.service;

import static java.util.stream.Collectors.toMap;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Lists;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import cn.hutool.core.util.IdUtil;
import lombok.extern.slf4j.Slf4j;
import xyz.xtt.common.starter.cache.CacheManager;
import xyz.xtt.schedule.dto.TblLuckPhonePoolDTO;
import xyz.xtt.schedule.entity.TblLuckPhonePool;
import xyz.xtt.schedule.mapper.equity.TblLuckPhonePoolMapper;
import xyz.xtt.schedule.support.GomeLuckPhoneHttpService;
import xyz.xtt.schedule.utils.RedisLockUtil;

/**
 * 靓号池服务
 *
 * @author dale
 * @since 2024/12/31
 **/
@Service
@Slf4j
public class LuckPhoneService extends ServiceImpl<TblLuckPhonePoolMapper, TblLuckPhonePool> implements DisposableBean {

	private static final String PHONE_POOL_BATCH_NO_CACHE_KEY = "NFT:CACHE:PHONE:POOL:BATCH";
	@Autowired
	private GomeLuckPhoneHttpService gomeLuckPhoneHttpService;
	@Autowired
	private TblRegionService tblRegionService;
	@Autowired
	private CacheManager cacheManager;
	@Autowired
	private RedisLockUtil redisLockUtil;
	@Autowired
	private StringRedisTemplate redisTemplate;

	private static final int FETCH_PHONE_PAGE_SIZE = 40;

	private static final int CORE_THREAD_COUNT = Math.min(Runtime.getRuntime().availableProcessors() * 2, 6);
	private static final ThreadFactory THREAD_FACTORY = new ThreadFactoryBuilder().setNamePrefix("CustomThread-%d").build();
	private static final ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(
			CORE_THREAD_COUNT, CORE_THREAD_COUNT, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(100), THREAD_FACTORY,
			new ThreadPoolExecutor.CallerRunsPolicy());

	private static final String PHONE_LOCK_STATUS = "1";
	private static final String PHONE_ON_SHELF = "1";
	private static final int HANDLE_PAGE_PER_TASK = 10;
	private static final String LOCK_KEY_REFRESH_PHONE_FMT = "NFT:LOCK:REFRESH:LUCK:PHONE";

	/**
	 * 同步手机号数据
	 */
	public void synPhoneData() {
		log.info("开始执行靓号池同步任务");
		String lockKey = LOCK_KEY_REFRESH_PHONE_FMT;
		String lockValue = UUID.randomUUID().toString();
		boolean locked = redisLockUtil.tryLock(lockKey, lockValue, 2, TimeUnit.MINUTES, 10, TimeUnit.SECONDS);
		if (locked) {
			try {
				// 本次同步批次号
				long batchNo = System.currentTimeMillis();

				// 获取总条数计算总页数
				Long totalCount = gomeLuckPhoneHttpService.fetchPhoneCount();
				if (totalCount == null || totalCount == 0) {
					log.info("synPhoneData: 无需同步");
					return;
				}

				int totalPage = (int) Math.ceil((double) totalCount / FETCH_PHONE_PAGE_SIZE);
				int taskCount = (int) Math.ceil((double) totalPage / HANDLE_PAGE_PER_TASK);
				log
						.info("synPhoneData: 开始同步，总条数：{}，每页处理条数：{}，总页数：{}, 线程数：{}, 子任务数：{}", totalCount, FETCH_PHONE_PAGE_SIZE, totalPage, EXECUTOR
								.getCorePoolSize(), taskCount);
				CountDownLatch latch = new CountDownLatch(taskCount);
				// 插入条数记录器
				AtomicInteger insertCounter = new AtomicInteger(0);
				for (int i = 1; i <= totalPage; i += HANDLE_PAGE_PER_TASK) {
					int pageNumFrom = i;
					int pageNumTo = i + HANDLE_PAGE_PER_TASK - 1;
					EXECUTOR.execute(() -> {
						try {
							synPhoneDataInner(pageNumFrom, pageNumTo, batchNo, insertCounter);
						} catch (Exception e) {
							log.error("synPhoneData: 处理异常 pageNumFrom:{} pageNumTo:{}", pageNumFrom, pageNumTo, e);
						} finally {
							latch.countDown();
							// 锁续期
							redisLockUtil.prolongLock(lockKey, lockValue, 2, TimeUnit.MINUTES);
						}
					});
				}
				try {
					latch.await();
				} catch (InterruptedException e) {
					log.error("synPhoneData: 等待异常", e);
					Thread.currentThread().interrupt();
				}
				log.info("synPhoneData: 处理完成，总条数：{}, 插入条数：{}", totalCount, insertCounter.get());
				if (insertCounter.get() > totalCount * 0.5) {
					updateCurrentBatchNo(batchNo);
				} else {
					log.error("插入条数较少，请排查同步过程是否出现异常");
				}
			} finally {
				redisLockUtil.releaseLock(lockKey, lockValue);
			}
		} else {
			log.info("refreshLuckPhone 获取锁失败");
		}
	}

	/**
	 * 更新当前批次号,并清除历史数据
	 * 
	 * @param batchNo
	 */
	private void updateCurrentBatchNo(long batchNo) {
		redisTemplate.opsForValue().set(PHONE_POOL_BATCH_NO_CACHE_KEY, String.valueOf(batchNo));
		QueryWrapper<TblLuckPhonePool> queryWrapper = new QueryWrapper();
		queryWrapper.lt("batch_no", batchNo);
		baseMapper.delete(queryWrapper);
	}

	/**
	 * 同步手机号数据
	 */
	private void synPhoneDataInner(int pageNoFrom, int pageNumTo, long batchNo, AtomicInteger insertCount) {
		List<TblLuckPhonePoolDTO> phoneList;
		do {
			phoneList = gomeLuckPhoneHttpService.fetchPhoneList(pageNoFrom++, FETCH_PHONE_PAGE_SIZE);
			handlePhoneList(phoneList, batchNo, insertCount);
		} while (!CollectionUtils.isEmpty(phoneList) && pageNoFrom <= pageNumTo);
	}

	private void handlePhoneList(List<TblLuckPhonePoolDTO> phoneList, long batchNo, AtomicInteger insertCount) {
		if (CollectionUtils.isEmpty(phoneList)) {
			return;
		}
		List<TblLuckPhonePool> needInsertList = new ArrayList<>();
		phoneList.forEach(phoneDTO -> {
			// 维护号码池
			if (!Objects.equals(phoneDTO.getLockStatus(), PHONE_LOCK_STATUS) && Objects.equals(phoneDTO.getShelfStatus(), PHONE_ON_SHELF)) {
				// 电话号码处于可出售状态需要检查对应省市区划数据是否存在,不存在则需要增加
				tblRegionService.ensureRegionDataExist(phoneDTO.getProvince(), phoneDTO.getCity());

				// 插入号码池
				phoneDTO.setId(IdUtil.getSnowflakeNextId());
				TblLuckPhonePool phoneEntity = new TblLuckPhonePool();
				this.copyProperties(phoneDTO, phoneEntity);
				phoneEntity.setBatchNo(batchNo);
				needInsertList.add(phoneEntity);
			}
		});

		try {
			if (!CollectionUtils.isEmpty(needInsertList)) {
				int count = this.insertBatch(needInsertList, 20);
				insertCount.addAndGet(count);
			}
		} catch (Exception e) {
			log.error("synPhoneData: 批量操作失败", e);
		}

	}

	private void copyProperties(TblLuckPhonePoolDTO phoneDTO, TblLuckPhonePool phoneEntity) {
		BeanUtils.copyProperties(phoneDTO, phoneEntity, "createTime", "delistDate", "listingDate", "batchNo");
		phoneEntity.setCreateTime(phoneDTO.convertDate(phoneDTO.getCreateTime()));
		phoneEntity.setDelistDate(phoneDTO.convertDate(phoneDTO.getDelistDate()));
		phoneEntity.setListingDate(phoneDTO.convertDate(phoneDTO.getListingDate()));
	}

	/**
	 * 查询已存在的号码
	 *
	 * @param phoneNumbers
	 * @param lastBatch
	 * @return
	 */
	private Map<String, TblLuckPhonePool> queryExistPhoneList(List<String> phoneNumbers, Long lastBatch) {
		QueryWrapper<TblLuckPhonePool> queryWrapper = new QueryWrapper<>();
		queryWrapper.in("tele_phone", phoneNumbers);
		if (lastBatch != null) {
			queryWrapper.eq("batch_no", lastBatch);
		}
		List<TblLuckPhonePool> resultList = baseMapper.selectList(queryWrapper);
		if (CollectionUtils.isEmpty(resultList)) {
			return Collections.emptyMap();
		}
		return resultList.stream().collect(toMap(TblLuckPhonePool::getTelephone, Function.identity(), (left, right) -> left));
	}

	@Override
	public void destroy() throws Exception {
		if (!EXECUTOR.isShutdown()) {
			EXECUTOR.shutdown();
		}
	}

	public int insertBatch(List<TblLuckPhonePool> dataList, int batchSize) {
		int insertCount = 0;
		List<List<TblLuckPhonePool>> partition = Lists.partition(dataList, batchSize);
		for (List<TblLuckPhonePool> list : partition) {
			insertCount += baseMapper.insertBatchIgnore(list);
		}
		return insertCount;
	}
}
